package com.huawei.dli.flink.sye.config;

import java.io.Serializable;
import java.sql.*;
import java.util.*;

/**
 * create table public.flink_ogg_meta_data
 * (
 * topic_name text, --topic名称
 * source_table_name text,--源端的表名称
 * hbase_table_name  text,--hbase的表名称
 * ogg_json text,--要取出的json字符串
 * row_key_column text,--选取rowkey的字段
 * status text --0代表正常使用 1 代表不使用
 * );
 */
public class FlinkOggMetaConfig implements Serializable {

    // 存储ogg解析源端的所有表 用户名.表名称
    public List<String> sourceTableNames = new ArrayList<>();
    // 存储源端表对应的rowkey
    public Map<String, String> rowkeysMap = new HashMap<>();
    // 存储目标端hbase的表名称
    public List<String> targetTableNames = new ArrayList<>();
    // 存储表对应解析的字段
    public Map<String, List<String>> tableColumnMap = new HashMap<>();
    // 存储原表和目标表的映射
    public Map<String, String> sourceTargetMap = new HashMap<>();

    /**
     * 根据topic解析出响应的配置
     *
     * @param topic
     */
    public void analyzeConfig(String topic) {
        String sql = "select source_table_name,hbase_table_name,ogg_json,row_key_column from public.flink_ogg_meta_data  " +
                " where topic_name='" + topic + "' and status=0";
        Connection conn = null;
        PreparedStatement ps = null;
        try {
            Class.forName("org.postgresql.Driver");
            conn = DriverManager.getConnection("jdbc:postgresql://25.34.21.154:8000/tjdl_sjzt_db1", "dbadmin", "Huawei@CLOUD8");
            ps = conn.prepareStatement(sql);
            ResultSet resultSet = ps.executeQuery();

            while (resultSet.next()) {
                String sourceTableName = resultSet.getString("source_table_name").trim();
                String hbaseTableName = resultSet.getString("hbase_table_name").trim();
                String oggJson = resultSet.getString("ogg_json").trim();
                String rowKeyColumn = resultSet.getString("row_key_column").trim();
                sourceTableNames.add(sourceTableName);
                rowkeysMap.put(sourceTableName, rowKeyColumn);
                targetTableNames.add(hbaseTableName);

                List<String> jsonColumn = Arrays.asList(oggJson.split(","));
                tableColumnMap.put(sourceTableName, jsonColumn);
                sourceTargetMap.put(sourceTableName, hbaseTableName);
            }
            System.out.println("总共查询到了 "+sourceTableNames.size()+" 个表的配置");
            System.out.println("配置的表为："+sourceTableNames);
        } catch (Exception e) {
            System.out.println("解析flink元数据：数据库连接异常");
            e.printStackTrace();
        } finally {
            try {
                if (null != conn) {
                    conn.close();
                }
                if (null != ps) {
                    ps.close();
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }
}